package state.inmemony;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

/**
 * @Author:RenPu
 * @Date: 2020/4/20 11:13
 * @Version: 1.0
 * @description:流数据处理器，单词分隔  line  ----->   word
 */
public class LineSplitBolt extends BaseRichBolt {



    //定义Bolt中的发射对象
    private OutputCollector collector;

    /**
     * 准备方法
     * @param map
     * @param topologyContext
     * @param outputCollector
     */
    @Override
    public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector outputCollector) {

    }


    /**
     * 执行方法，Bolt每接受一个Tuplt，都会调用此方法
     * @param tuple
     */

    @Override
    public void execute(Tuple tuple) {


        //通过上一个Spolt处理器的唯一标识，获取所对应的数据
        String line = tuple.getStringByField("line");

        //设定字符串分隔的规律以空格分隔
        String[] word = line.split(" ");

        //遍历同时将数据封装到元祖对象，进行发射到下一个处理器
        for (String words: word) {

            //新建元祖，交由下游处理器进行统计和计算

            /**
             * tuple锚定父tuple，并且维护tuple树
             */
            collector.emit(tuple,new Values(words));


            /**
             * 应答方法：表示处理器已经完成对父tuple处理完成
             */
            collector.ack(tuple);
        }
        


    }


    /**
     *
     * @param declarer
     */

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

        //声明元祖的唯一标识，供下一组件获取里面的数据
        declarer.declare(new Fields("word"));

    }
}
